package net.bwie.realtime.dwd.douyin.logs.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dwd.douyin.logs.bean.EventLog;
import net.bwie.realtime.dwd.douyin.logs.bean.TradePay;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import net.bwie.realtime.jtp.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

public class EventPayAssociation {
    public static void main(String[] args) throws Exception{
        //1.执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.数据源-source
        DataStream<String> dwd_event_log = KafkaUtil.consumerKafka(env, "dwd_event_log");
        DataStream<String> dwd_trade_pay_detail = KafkaUtil.consumerKafka(env, "dwd_trade_pay");


        //3.数据转换-transformation
        DataStream<String> pageStream = processLog(dwd_event_log, dwd_trade_pay_detail);
        pageStream.print();

        //5.触发执行-execute
        env.execute("EventPayAssociation");
    }

    private static DataStream<String> processLog(DataStream<String> dwd_event_log,
                                                 DataStream<String> dwd_trade_pay_detail) {
        SingleOutputStreamOperator<EventLog> eventLogProcess = eventLog(dwd_event_log);

        SingleOutputStreamOperator<TradePay> tradePayProcess = tradePay(dwd_trade_pay_detail);
        DataStream<String> handleStream = handle(eventLogProcess, tradePayProcess);
//        handleStream.print();

        KafkaUtils.producerKafka(handleStream,"dws_event_pay_transaction");

        return handleStream;
    }

    private static DataStream<String> handle(SingleOutputStreamOperator<EventLog> eventLogProcess,
                                             SingleOutputStreamOperator<TradePay> tradePayProcess) {
        // 1. 将事件流按照主播ID分组
        KeyedStream<EventLog, Integer> eventKeyedByUserId = eventLogProcess
                .keyBy(eventLog -> eventLog.getAnchorId());

        // 2. 将支付流按照主播ID分组
        KeyedStream<TradePay, Integer> payKeyedByUserId = tradePayProcess
                .keyBy(tradePay -> tradePay.getAnchorId());

        DataStream<String> joinedStream = payKeyedByUserId
                .intervalJoin(eventKeyedByUserId)
                .between(Time.seconds(-30), Time.seconds(30)) // 支付前30分钟到支付后30分钟
                .process(new ProcessJoinFunction<TradePay, EventLog, String>() {
                    @Override
                    public void processElement(TradePay pay, EventLog event, Context ctx, Collector<String> out) {
                        // 4. 处理关联结果，这里将结果转换为JSON字符串
                        Map<String, Object> result = new HashMap<>();
                        result.put("pay_info", pay);
                        result.put("event_info", event);
                        result.put("association_time", new Date());

                        out.collect(JSON.toJSONString(result));
                    }
                });
        SingleOutputStreamOperator<String> filter = joinedStream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String payStatus = jsonObject.getJSONObject("pay_info").getString("payStatus");
                if (payStatus.equals("success")) {
                    return true;
                } else {
                    return false;
                }
            }
        });

        return filter;
    }

    private static SingleOutputStreamOperator<TradePay> tradePay(DataStream<String> stream) {
        SingleOutputStreamOperator<TradePay> eventLog = stream.map(new MapFunction<String, TradePay>() {
            @Override
            public TradePay map(String s) throws Exception {
                return JSON.parseObject(s, TradePay.class);
            }
        });
        SingleOutputStreamOperator<TradePay> tradepayStream = eventLog.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradePay>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TradePay>() {
                            @Override
                            public long extractTimestamp(TradePay tradePay, long l) {
                                String enterTime = tradePay.getPayTime();
                                Long aLong = DateTimeUtil.convertStringToLong(enterTime, "yyyy-MM-dd HH:mm:ss");
                                return aLong;
                            }
                        })
        );

        return tradepayStream;
    }

    private static SingleOutputStreamOperator<EventLog> eventLog(DataStream<String> stream) {
        SingleOutputStreamOperator<EventLog> eventLog = stream.map(new MapFunction<String, EventLog>() {
            @Override
            public EventLog map(String s) throws Exception {
                return JSON.parseObject(s, EventLog.class);
            }
        });
        SingleOutputStreamOperator<EventLog> eventLogStream = eventLog.assignTimestampsAndWatermarks(
                WatermarkStrategy.<EventLog>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                        .withTimestampAssigner(new SerializableTimestampAssigner<EventLog>() {
                            @Override
                            public long extractTimestamp(EventLog eventLog, long l) {
                                String enterTime = eventLog.getEnterTime();
                                Long aLong = DateTimeUtil.convertStringToLong(enterTime, "yyyy-MM-dd HH:mm:ss");
                                return aLong;
                            }
                        })
        );

        return eventLogStream;
    }


}
